package com.etlmaster.executor.steps.control;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import com.etlmaster.executor.EtlMasterException;
import com.etlmaster.executor.Job;
import com.etlmaster.executor.JobPool;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.StepStack;
import com.etlmaster.executor.utils.CallableResult;
import com.etlmaster.executor.utils.LogWriter;
import com.etlmaster.executor.utils.MultiProcessHelper;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Larry
 * @date 2017-04-14 循环接口
 */
public abstract class LoopStepInterface extends StepMeta {

	private int threadNum;
	public LoopStepInterface(String id, String name, String step_code,
			String session_id, String paramStr) {
		super(id, name, step_code, session_id, paramStr);
		stack=StepStack.stepStacks.get(id+name);
		if(stack==null){
			stack=new StepStack(id+name+"_"+session_id,getThreadNum(),true);
			StepStack.stepStacks.put(id+name, stack);
		}
	}

	public void loop(StepStack stack, StepMeta curStepMeta, Job job) {
		this.getStepState().setStatus(StepState.RUNNING);
		int rtn = process(stack, curStepMeta, job);
		this.getStepState().setStatus(rtn);
	}

	private Future<CallableResult> future;

	private StepStack stack;
	public abstract int process(StepStack stack, StepMeta curStepMeta, Job job);

	@Override
	protected void loadParamsFromConfigTable() {

	}
	public void runLoop( Job job, int i,String param) throws InterruptedException, ExecutionException {

		Job jobinstance = JobPool.getJob(job.getMeta().getName());
		JobPool.copyVariables(job.getSeesionId(), jobinstance.getSeesionId());

		jobinstance.putVariable(param, i);
		jobinstance.setTmpJob(true);
		StepMeta meta = getLoopNextStep(jobinstance);
		stack.pushStack(stack.getHelper().getWaitQueue(), meta, jobinstance,
				job.getSeesionId() + "_" + i);

		if (future==null&&i==this.getThreadNum()) {
			future = MultiProcessHelper.executor.submit(stack.getStackThread());
		}else if (future!=null&&future.isDone()) {
			CallableResult result =future.get();
			if (result != null) {
				if (result.isError()) {
					throw new EtlMasterException(result.getMessage());
				}
			}
		}
	}

	public void startLoopStack() throws ExecutionException {

		try {
			if (future==null) {
				future = MultiProcessHelper.executor.submit(stack.getStackThread());
			}
			stack.setAllFetched(true);
			CallableResult result =future.get();
			if (result != null) {
				if (result.isError()) {
					throw new EtlMasterException(result.getMessage());
				}
			}
			future=null;
		} catch (InterruptedException e) {
			this.log("ERROR", e.getMessage());
		}
	}

	public StepMeta getLoopNextStep(Job job) {
		List<StepMeta> result = job.getNextStepMetas(this.getId());
		return result.get(0);
	}

	public List<StepMeta> getEndLoopNextSteps(Job job) {
		List<StepMeta> result = job.getNextStepMetas(this.getId());
		result.remove(0);
		return result;
	}

	public StepStack getStack() {
		return stack;
	}

	public void setStack(StepStack stack) {
		this.stack = stack;
	}

	public int getThreadNum() {
		if(!(threadNum>0)){
			return 1;
		}
		return threadNum;
	}

	public void setThreadNum(int threadNum) {
		this.threadNum = threadNum;
	}
}
